#include <iostream>
#include <string>
#include <vector>

#include <google/protobuf/descriptor.h>
#include <google/protobuf/compiler/importer.h>
#include <google/protobuf/dynamic_message.h>

#include <qpid/client/Connection.h>
#include <qpid/client/Session.h>
#include <qpid/client/Message.h>
#include <qpid/client/SubscriptionManager.h>
#include <qpid/client/LocalQueue.h>

#include <boost/program_options.hpp>
#include <boost/format.hpp>
#include <boost/foreach.hpp>

using namespace std;
using namespace google;
using namespace qpid;
namespace po = boost::program_options;

#define foreach BOOST_FOREACH

string getMessageName(const string &strRoutingKey)
{
   size_t pos = strRoutingKey.find('.');
   size_t pos2 = strRoutingKey.find('.', pos+1);
   string msgName = strRoutingKey.substr(pos+1, pos2-pos);

   pos = strRoutingKey.rfind('.');
   msgName += strRoutingKey.substr(pos+1);

   return msgName;
}

class ProtoLoaderErrorCollector : public protobuf::compiler::MultiFileErrorCollector {
   public:
      ProtoLoaderErrorCollector() { };
      virtual ~ProtoLoaderErrorCollector() { };
      virtual void AddError(const string &filename,
                              int line,
                              int column,
                              const string &message)
      {

         cerr << "Error loading proto:" << filename << endl
              << message << endl;
      }
};

const char *gv_program = "readBroker";
const char *MY_SQROOT = "MY_SQROOT";
const char *QPID_NODE_PORT = "QPID_NODE_PORT";

int main(int argc, char** argv) {

   string brokerIp;
   size_t brokerPort;
   string address = "amq.topic";
   string protoSrc = "/h/caiche/Programs/protoSrc";
   vector<string> bindingKeys;

   char *pSqRoot = getenv(MY_SQROOT);
   if (pSqRoot) {
      protoSrc = pSqRoot;
      protoSrc += "/seapilot/export/publications";
   }

   po::options_description opts("Allowed options");
   opts.add_options()
        ("help,h", "produce this help message")
        ("broker-ip", po::value<string>(&brokerIp)->default_value("127.0.0.1"), 
                              "broker ip")
        ("broker-port", po::value<size_t>(&brokerPort)->default_value(64790), 
                              "broker port")
        ("binding-key", po::value< vector<string> >(&bindingKeys), "keys to bind")
        ("proto-src", po::value<string>(&protoSrc)->default_value(protoSrc), "root directory of publications")
       ;

   po::variables_map vm;
   try {
      po::store(po::parse_command_line(argc, argv, opts), vm);
      po::notify(vm);
   }
   catch ( const po::error& e ) {
      cerr << e.what() << endl;
      exit(-1);
   }

   if (vm.count("help")) {
        cout << "Usage: " << gv_program << " [options]" << endl;
        cout << opts;
        return 0;
   }

   if (bindingKeys.empty())
      bindingKeys.push_back("#");

   client::Connection conn;
   try {
      conn.open(brokerIp.c_str(), brokerPort);
      client::Session session = conn.newSession();


      //declare queue
      string queueName = "get_queue";
      session.queueDeclare(client::arg::queue = queueName, 
                           client::arg::autoDelete = true,
                           client::arg::exclusive = false);

      // bind keys
      string address = "amq.topic";
      foreach(const string &key, bindingKeys) {
         cout << "binding " << key << endl;
         session.exchangeBind(client::arg::exchange = address,
                              client::arg::queue = queueName,
                              client::arg::bindingKey = key);
      }
      
      // subscribeQueue
      client::SubscriptionManager subscriptions(session);
      client::LocalQueue localQueue;
      subscriptions.subscribe(localQueue, queueName);

      // listen
      ProtoLoaderErrorCollector errorCollector;
      protobuf::compiler::DiskSourceTree sourceTree;
      protobuf::compiler::Importer importer(&sourceTree, &errorCollector);
      protobuf::DynamicMessageFactory messageFactory(importer.pool());

      sourceTree.MapPath("", protoSrc);

      while(true) {
         client::Message message;

         localQueue.get(message, sys::TIME_INFINITE);

         string strRoutingKey = message.getDeliveryProperties().getRoutingKey();

         cout << endl << "Message: " << strRoutingKey << endl;

         string pubName = getMessageName(strRoutingKey);
         const protobuf::Descriptor *desc = 
            importer.pool()->FindMessageTypeByName(pubName);

         if(NULL == desc) // probably haven't imported it if we don't have it
         {
            string filename = pubName + ".proto";
            importer.Import(filename);
            desc = importer.pool()->FindMessageTypeByName(pubName);
         }

         if(NULL == desc) {
            cout << "Wrong key: " << strRoutingKey << endl;
            break;
         }

         protobuf::Message *m =  messageFactory.GetPrototype(desc)->New();

         m->ParseFromString(message.getData());
         cout << "{" << endl
              << m->DebugString()
              << "}" << endl;

      }

      conn.close();
      return 0;

   } 
   catch(const exception& error) {
         cerr << error.what() << endl;
         conn.close();
         return 1;   
   }

}
